package com.hailiang.reptile;

import com.hailiang.scheduler.UrlSchedulerImp;
import org.apache.commons.configuration.ConfigurationFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.htmlcleaner.HtmlCleaner;
import org.htmlcleaner.TagNode;
import org.htmlcleaner.XPatherException;

import java.io.IOException;
import java.net.StandardSocketOptions;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    static HtmlCleaner htmlCleaner = new HtmlCleaner();
    static UrlSchedulerImp urlSchedulerImp = new UrlSchedulerImp();
    static UrlSchedulerImp urlPullSchedulerImp = new UrlSchedulerImp();
    public static void main(String[]args) throws Exception {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    new Reptile().reptile("https://tieba.baidu.com/f?kw=%E7%BE%8E%E5%A5%B3&ie=utf-8&pn=");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        Runnable runnable1 = new Runnable() {
            @Override
            public void run() {
                Reptile reptile = new Reptile();
                try {
                    Thread.sleep(3000);
                    //创建hbase 表   并保存在里面
                    Configuration conf = HBaseConfiguration.create();
                    conf.set("hbase.zookeeper.quorum","uplook03:2181,uplook04:2181,uplook05:2181");
                    Connection conn = ConnectionFactory.createConnection(conf);
                    Table table = conn.getTable(TableName.valueOf("ns2:t1"));
                    while (true){
                        String url = urlSchedulerImp.pull("url");
                        String html = reptile.downLoad("https://tieba.baidu.com" + url);
                        TagNode tagNode = htmlCleaner.clean(html);
                        String title = tagNode.evaluateXPath("//h3/text()")[0].toString();
                        urlSchedulerImp.push("title",title);

                        Object[] objects = tagNode.evaluateXPath("//img[@class='BDE_Image']/@src");
                        for (Object object : objects) {
                            System.out.println(object);
                            urlSchedulerImp.push("photo",object.toString());
                        }
                        /**
                         * 数据存入hbase 中   rowkey设计  md5 标题+时间戳 列设计 f1 title f2 ingUrl
                         */
                        String photo = urlPullSchedulerImp.pull("photo");
                        long time = System.currentTimeMillis();
                        String md5String = Md5Util.getMd5String(String.valueOf(time) + title);
                        String rowKey = md5String+","+title+time;
                        Put put = new Put(Bytes.toBytes(rowKey));
                        put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("title"), Bytes.toBytes(title));
                        put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("imageUrls"), Bytes.toBytes(photo));
                        table.put(put);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.submit(runnable);
        executorService.submit(runnable1);
    }
}
